#include "db_nodata.h"
struct db* dbopen(char *name,int flag){
    struct hparam hp;
	memset(&hp,0,sizeof(struct hparam));
	hp.bnum=65535;
	hp.minkey=2;
	hp.pgsiz=512;
	hp.cache_maxbytes=1024*64;
	hp.cache_nbucket=61;
	if(name==NULL){ 
        fprintf(stderr,"Can't create an anonymous database\n");
		return NULL;
	}
	struct db *gdb=calloc(1,sizeof(struct db));
	if(gdb==NULL){
	    fprintf(stderr,"Memory allocate error\n");
		return NULL;
	}
	int nlen=strlen(name);
	int slen=strlen(DB_FILE_SUFFIX);
	if(!endwith(name,nlen,DB_FILE_SUFFIX,slen)){
	    gdb->dname=malloc(nlen+slen+1);
		if(gdb->dname==NULL){
		    fprintf(stderr,"Memory allocate error\n");
			free(gdb);
			return NULL;
		}
		memcpy(gdb->dname,name,nlen);
		memcpy(gdb->dname+nlen,DB_FILE_SUFFIX,slen);
		gdb->dname[nlen+slen]='\0';
		nlen+=slen;
	}else{
	    gdb->dname=malloc(nlen);
		memcpy(gdb->dname,name,nlen);
	}
	int ilen=nlen+8+strlen(DEF_BDB_SUFFIX)+1;
	char *iname=malloc(ilen);
	if(iname==NULL){
		fprintf(stderr,"Mem error\n");
	    free(gdb);
	}
	snprintf(iname,ilen,"%s.[0-9]*.%s",gdb->dname,DEF_BDB_SUFFIX);
	DEBUG(printf("%s:dbname=%s iname=%s\n",__func__,gdb->dname,iname));
	if(access(gdb->dname,F_OK)!=0){
		glob_t ifiles;
		memset(&ifiles,0,sizeof(ifiles));
		DEBUG(printf("%s:file already exsiting\n",__func__));
		if(glob(iname,GLOB_ERR|GLOB_NOSORT,NULL,&ifiles)==0){
		    int i;
			for(i=0;i<ifiles.gl_pathc;i++){
			    DEBUG(printf("%s:find exsit idx file:%s\n",__func__,ifiles.gl_pathv[i]));
			    if(unlink(ifiles.gl_pathv[i])<0){
			        perror("unlink");
					free(gdb->dname);
					free(gdb);
					free(iname);
					return NULL;
			    }
			}
		}
		globfree(&ifiles);
	}
	
	if((gdb->hd=hdbopen(gdb->dname,flag,&hp))==NULL){
	    fprintf(stderr,"Can't open data file\n");
		free(gdb->dname);
		free(gdb);
		free(iname);
		return NULL;
	}
	glob_t ifiles;
	memset(&ifiles,0,sizeof(ifiles));
	if(glob(iname,GLOB_ERR|GLOB_NOSORT,NULL,&ifiles)==0){
	    struct bdb *idb;
		int i;
		struct bstring *col;
		gdb->cdbs=hashcreate(0);
		for(i=0;i<ifiles.gl_pathc;i++){
		    char *ptr=ifiles.gl_pathv[i]+nlen+1;
			int seq=atoi(ptr);
			if(gdb->seq<seq) gdb->seq=seq;
		    if((idb=bopen(gdb->dname,NULL,seq,flag,NULL))!=NULL){
		        col=colname(idb,NULL);
		        hashput(gdb->cdbs,col->array,col->siz,(void*)idb);
				strdel(col);
		    }else{
		        printf("bad index file:%s\n",ifiles.gl_pathv[i]);
		    }
		}
		globfree(&ifiles);
	}
	free(iname);
	gdb->flag=flag;
	pthread_rwlock_init(&gdb->rwlock,NULL);
	//pthread_mutex_init(&gdb->lock,NULL);
	return gdb;
}
void dbclose(struct db *gdb){
    //pthread_mutex_lock(&gdb->lock);
	pthread_rwlock_wrlock(&gdb->rwlock);
    hdbclose(gdb->hd);
	if(gdb->cdbs!=NULL){
	    hashdel(gdb->cdbs,(void(*)(void*))bclose);
	}
	free(gdb->dname);
	pthread_rwlock_unlock(&gdb->rwlock);
	pthread_rwlock_destroy(&gdb->rwlock);
	//pthread_mutex_destroy(&gdb->lock);
	free(gdb);
}
int simpleput(struct db *gdb,char* id,int idsiz,char *rec[]){
    int i=0,ret=true;
	char buf[BLOCKSIZ];
	char* val=buf;
	char *ptr=val;
	int siz=BLOCKSIZ;
	struct bdb *idb;
	int cursiz,ksiz,vsiz;
#define SIMPLECHECK(need)\
    do{\
	    cursiz=ptr-val;\
		if(siz-cursiz<(need)){\
		    if(val==buf){\
			    siz<<=1;\
			    val=malloc(siz);\
				memcpy(val,buf,cursiz);\
				ptr=cursiz+val;\
			}else{\
			    siz<<=1;\
				val=realloc(val,siz);\
				ptr=val+cursiz;\
			}\
		}\
	}while(0)
	pthread_rwlock_rdlock(&gdb->rwlock);
	while(rec[i]!=NULL){
	    SIMPLECHECK(sizeof(int32));
	    ksiz=*((int32*)ptr)=strlen(rec[i]);
		ptr+=sizeof(int32);
		SIMPLECHECK(ksiz);
		memcpy(ptr,rec[i],ksiz);
		ptr+=ksiz;
		SIMPLECHECK(sizeof(int32));
		if(rec[i+1]!=NULL){
		    vsiz=*((int32*)ptr)=strlen(rec[i+1]);
		    ptr+=sizeof(int32);
		    SIMPLECHECK(vsiz);
			memcpy(ptr,rec[i+1],vsiz);
		}else{
		    *((int32*)ptr)=0;
		}
		i+=2;
	}
	if(id!=NULL||idsiz!=0){
	    if(!hdbput(gdb->hd,id,idsiz,val,ptr-val,0)){
		    printf("%s:error put rec\n",__func__);
			ret=false;
		}
	}
	else if(!hdbput(gdb->hd,(char*)&gdb->hd->Lsn,sizeof(gdb->hd->Lsn),val,ptr-val,0)){
	    printf("put fail\n");
		if(val!=buf) free(buf);
		pthread_rwlock_unlock(&gdb->rwlock);
		return false;
	}else gdb->hd->Lsn++;
	if(val!=buf) free(buf);
	i=0;
	if(gdb->cdbs!=NULL){
	    while(rec[i]!=NULL){
	        if((idb=hashget(gdb->cdbs,rec[i],strlen(rec[i])))!=NULL
			&&!bput(idb,rec[i+1],rec[i+1]==NULL?0:strlen(rec[i+1]),(char*)&gdb->hd->Lsn,sizeof(gdb->hd->Lsn),0)){
			    printf("put idx fail\n");
				ret=false;
			}
	    }
	}
	pthread_rwlock_unlock(&gdb->rwlock);
	return ret;
}
int dbputstr(struct db *gdb,char *id,int idsiz,char *str,int siz){
    char *delim=" \t\n\r";
	int ssiz=10;
	char *buf[10];
	char **stack=buf;
	int top=0;
	char *pre=str;
	char *next=pre;
	int dsiz=strlen(delim);
#define STRSTACKPUSH(ptr)\
    do{\
	    if(top>=ssiz){\
		   ssiz<<=1;\
		   if(stack==buf){\
		        stack=malloc(ssiz*sizeof(*stack));\
				memcpy(stack,buf,sizeof(buf));\
			}\
		    else stack=realloc(stack,ssiz*sizeof(*stack));\
		}\
		stack[top++]=(ptr);\
	}while(0)
	while(pre<str+siz){
	    while(memchr(delim,*next,dsiz)==NULL)
	        next++;
		if(((top&1)==0&&next-pre!=0)||(top&1)==1)
		    STRSTACKPUSH(pre);
		next++;
		pre=next;
	}
	if((top&1)==0){
	    printf("%s:bad parameter\n",__func__);
		if(stack!=buf) free(stack);
		return false;
	}
	STRSTACKPUSH(NULL);
	int ret=simpleput(gdb,id,idsiz,stack);
	if(stack!=buf) free(stack);
	return ret;
}
int dbput(struct db *gdb,char *id,int idsiz,struct harray *ha){
	struct bdb *idb;
	struct hdb *hd=gdb->hd;
	struct bstring *key;
	int ret=1;
	struct item *it=ha->first;
	if(ha->nrec==0){
	    return false;
	}
	pthread_rwlock_rdlock(&gdb->rwlock);
	key=strnew(BLOCKSIZ);
	while(it!=NULL){
	    if(gdb->cdbs!=NULL&&(idb=(struct bdb*)hashget(gdb->cdbs,it->key->array,it->key->siz))!=NULL){
		    DEBUG(printf("%s:key=%s\n",__func__,it->key->array));
			struct bstring *val=it->val;
		    if(!bput(idb,val->array,val->siz,(char*)&hd->Lsn,sizeof(hd->Lsn),0)){
			    printf("%s:error put %s-%s\n",__func__,it->key->array,val->array);
				ret=false;
			}
		}
		stradd(key,(char*)&it->key->siz,sizeof(it->key->siz));
		stradd(key,it->key->array,it->key->siz);
		struct bstring *val=it->val;
		stradd(key,(char*)&val->siz,sizeof(val->siz));
		stradd(key,val->array,val->siz);
		DEBUG(printf("%s:key->siz=%d\n",__func__,key->siz));
		it=it->next;
	}
	if(id!=NULL||idsiz!=0){
	    if(!hdbput(hd,id,idsiz,key->array,key->siz,0)){
		    printf("%s:put data fail\n",__func__);
			pthread_rwlock_unlock(&gdb->rwlock);
			ret=false;
		}
	}
	else if(!hdbput(hd,(char*)&hd->Lsn,sizeof(hd->Lsn),key->array,key->siz,0)){
	    printf("%s:put data fail\n",__func__);
		ret=false;
	}else hd->Lsn++;
	strdel(key);
	pthread_rwlock_unlock(&gdb->rwlock);
	return ret;
}
int dbputrec(struct db *gdb,char *id,int idsiz,char *arg[]){
    struct harray* ha=hashcreate(0);
	int i=0;
	while(arg[i]!=NULL){
	    struct bstring *k=strcreate(arg[i],strlen(arg[i]));
		struct bstring *v=strcreate(arg[i+1],strlen(arg[i+1]));
		DEBUG(printf("%s:key=%s val=%s\n",__func__,arg[i],arg[i+1]));
		hashputkeep(ha,k,v);
		i+=2;
	}
	int ret=dbput(gdb,id,idsiz,ha);
	hashdelstr(ha);
	return ret;
}
int checkflag(struct condition *cond){
    if(cond->flag&C_NG){
	    cond->flag&=~C_NE;
	    if(cond->flag&C_EQ){
		    cond->flag&=~C_EQ;
			cond->flag|=C_NE;
		}else if(cond->flag&C_LS){
		    cond->flag&=~C_LS;
			cond->flag|=C_GE;
		}else if(cond->flag&C_LE){
		    cond->flag&=~C_LE;
			cond->flag|=C_GR;
		}else if(cond->flag&C_GR){
		    cond->flag&=~C_GR;
			cond->flag|=C_LE;
		}else if(cond->flag&C_GE){
		    cond->flag&=~C_GE;
			cond->flag|=C_LS;
		}else if(cond->flag&C_NE){
		    cond->flag&=~C_NE;
			cond->flag|=C_EQ;
		}else{
		    printf("%s:bad flag=%x\n",__func__,cond->flag);
			return false;
		}
	}
	return true;
}
struct harray* query(struct db *gdb,struct condition *c,int max,int skip,struct order *ord){
    struct bdb *idb=NULL;
	if(gdb->cdbs!=NULL)	idb=hashget(gdb->cdbs,c->key,c->ksiz);
	struct harray *ha;
	if(!checkflag(c)) return NULL;
	if(idb==NULL){
	    ha=hashcreate(0);
	    struct iterator *iter=iter_create(gdb->hd);
		struct bstring *key,*val;
		while(iter_next(iter,&key,&val)){
		    struct harray *rec=strtohash(val->array,val->siz);
			if(!checkcond(rec,c)){
			    strdel(key);
			    hashdelstr(rec);
			}
			else hashputkeep(ha,key,rec);
			strdel(val);
		}
		iterdel(iter);
	}else{
	    if((ha=condfind(gdb,idb,c))==NULL||ha->nrec==0){
		    if(ha!=NULL) hashdel(ha,NULL);
			return NULL;
		}
		struct item *it=ha->first;
		struct bstring *val;
		while(it!=NULL){
		    val=hdbgetstr(gdb->hd,it->key);
			DEBUG(printf("%s:val=%s\n",__func__,val->array));
			if(val!=NULL&&val->siz>0){
			    struct harray *rec=strtohash(val->array,val->siz);
			    it->val=rec;
			}
			strdel(val);
			it=it->next;
		}
	}
	if(ord!=NULL) hashsort(ha,ord);
	if(skip!=0){
		hashheadout(ha,skip,((void(*)(void*))hashdelstr));
		if(ha->nrec==0) hashdel(ha,NULL);
		return NULL;
	}
	if(max>0&&ha->nrec>max){
	    hashtailout(ha,ha->nrec-max,((void(*)(void*))hashdelstr));
	}
	return ha;
}
struct harray* condfind(struct db *gdb,struct bdb *idb,struct condition *cond){
	struct harray *ha;
	if(cond->flag&C_EQ){
	    DEBUG(printf("%s:ref=%s\n",__func__,cond->ref));
		bgettohash(idb,cond->ref,cond->rsiz,&ha);
	}else if((cond->flag&C_GR)||(cond->flag&C_GE)){
	    ha=hashcreate(0);
	    biterinit(idb,cond->ref,cond->rsiz,cond->flag&C_GE?BI_INCLUDE:0);
		struct bstring *key=NULL;
		struct bstring *val=NULL;
		while(biternext(idb,&key,&val)){
		    hashputkeep(ha,val,NULL);
			//strdel(key);
			val=NULL;
		}
		if(key!=NULL) strdel(key);
		if(val!=NULL) strdel(val);
	}else if((cond->flag&C_LS)||(cond->flag&C_LE)){
	    biterinit(idb,cond->ref,cond->rsiz,cond->flag&C_LE?BI_INCLUDE|BI_BACKWORD:BI_BACKWORD);
		struct bstring *key=NULL;
		struct bstring *val=NULL;
		ha=hashcreate(0);
		while(biternext(idb,&key,&val)){
		    hashputkeep(ha,val,NULL);
			//strdel(key);
			val=NULL;
		}
		if(key!=NULL) strdel(key);
		if(val!=NULL) strdel(val);
	}else if(cond->flag&C_NE){
	    biterinit(idb,NULL,0,0);
		struct bstring *key=NULL;
		struct bstring *val=NULL;
		ha=hashcreate(0);
		while(biternext(idb,&key,&val)){
		    if(strcompare(key,cond->key,cond->ksiz)!=0){
		        hashputkeep(ha,val,NULL);
				val=NULL;
			}/*else{
				strdel(val);
			}
			strdel(key);*/
		}
		if(key!=NULL) strdel(key);
		if(val!=NULL) strdel(val);
	}else{
		return NULL;
	}
	return ha;
}
int checkcond(struct harray *ha,struct condition *c){
    struct bstring *val;
    if((val=hashget(ha,c->key,c->ksiz))==NULL) return true;
	int cmp=strncmp(val->array,c->ref,c->rsiz>val->siz?val->siz:c->rsiz);
	if(cmp==0) cmp=val->siz-c->rsiz;
	if(((c->flag&C_GR)&&cmp>0)
	   ||((c->flag&C_GE)&&cmp>=0)
	   ||((c->flag&C_LS)&&cmp<0)
	   ||((c->flag&C_LE)&&cmp<=0)
	   ||((c->flag&C_EQ)&&cmp==0)
	   ||((c->flag&C_NE)&&cmp!=0))
	    return true;
	else return false;
}
int checkconds(struct db *gdb,struct harray *ha,struct condition *conds,int nc){
    int i;
	struct item *it=ha->first;
	struct item *tmp;
	struct bstring *val;
	struct harray *h;
	while(it!=NULL){
	    h=it->val;
		tmp=it->next;
	    if(h==NULL){
	        val=hdbgetstr(gdb->hd,it->key);
		    if(val!=NULL){
		        it->val=h=strtohash(val->array,val->siz);
				strdel(val);
			}
		}
		if(h!=NULL){
		    for(i=0;i<nc;i++)
		        if(!(conds[i].flag&C_DEAD)&&!checkcond(h,&conds[i])){
				    hasherase(ha,it->key->array,it->key->siz);
					hashdelstr(h);
					printf("%s:rec %d check fail\n",__func__,i);
					break;
			    }
		}else{//can't happen here
		    hasherase(ha,it->key->array,it->key->siz);
		}
		it=tmp;
	}
	return true;
}
struct harray* dbget(struct db *gdb,struct condition *conds,int nc,int max,int skip,struct order *ord){
	struct harray *ha1=NULL;
	struct harray *ha2=NULL;
	if(nc==0) return NULL;
	else if(nc==1){
	    return query(gdb,conds,max,skip,ord);
	}
	struct bdb *idb;
	int i,k=0;
	for(i=0;i<nc&&checkflag(&conds[i]);i++);//modify the negative flag
	if(i<nc) return NULL;
	for(i=0;i<nc;i++){//and
	    idb=NULL;
	    if(((idb=hashget(gdb->cdbs,conds[i].key,conds[i].ksiz))!=NULL)
		   &&!(conds[i].flag&C_NE)
		   &&k<2){
		    idb=hashget(gdb->cdbs,conds[i].key,conds[i].ksiz);
			if(ha1==NULL)
		        ha1=condfind(gdb,idb,&conds[i]);
			else{
			    ha2=condfind(gdb,idb,&conds[i]);
				hashand(ha1,ha2);
				hashdel(ha2,NULL);
			}
			DEBUG(printf("%s:find %d rec field=%s\n",__func__,ha1->nrec,conds[i].key));
		    conds[i].flag|=C_DEAD;
			k++;
		}
	}
	if(ha1==NULL) ha1=hashcreate(0);
	//printf("%s:and got %d rec\n",__func__,ha1->nrec);
	for(i=0;i<nc;i++){//or
	    idb=NULL;
	    if(!(conds[i].flag&C_DEAD)&&(conds[i].flag&C_OR)){
		    if((idb=hashget(gdb->cdbs,conds[i].key,conds[i].ksiz))!=NULL){
			    ha2=condfind(gdb,idb,&conds[i]);
			}else{
			    ha2=hashcreate(0);
				struct iterator *it=iter_create(gdb->hd);
				struct bstring *key,*val;
				while(iter_next(it,&key,&val)){
				    struct harray *h=strtohash(val->array,val->siz);
					hashputkeep(ha2,key,h);
					strdel(val);
				    strdel(key);
				}
				iterdel(it);
			}
			struct item *it=ha2->first;
			while(it!=NULL){
			    hashputrep(ha1,it->key->array,it->key->siz,it->val);
				it=it->next;
			}
			hashdel(ha2,(void(*)(void*))hashdelstr);
			conds[i].flag|=C_DEAD;
		}
	}
	//printf("%s:before check\n",__func__);
	//hashtraverse(ha1);
	checkconds(gdb,ha1,conds,nc);
	if(ord!=NULL)
	    hashsort(ha1,ord);
	//printf("%s:got %d rec ha1->first=%d\n",__func__,ha1->nrec,*((uint64*)(ha1->first->key->array)));
	//hashtraverse(ha1);
	if(skip!=0){
	    hashheadout(ha1,skip,(void(*)(void*))hashdelstr);
	}
	
	if(max!=0&&ha1->nrec>max){
	    hashtailout(ha1,ha1->nrec-max,(void(*)(void*))hashdelstr);
	}
	return ha1;
}
void hashsort(struct harray *ha,struct order *ord){
    struct item *it=ha->first;
	if(it==NULL||ha->nrec<=1) return;
	struct sortitem *s=malloc(sizeof(struct sortitem)*ha->nrec);
	int i=0;
	while(it!=NULL&&i<ha->nrec){
	    struct harray *rec=(struct harray*)it->val;
		struct bstring *val=hashget(rec,ord->key,ord->ksiz);
		if(val==NULL){
		    printf("%s:value is NULL.sortkey=%s ksiz=%d\n",__func__,ord->key,ord->ksiz);
			exit(1);
		}
		s[i].key=val;
		s[i].data=it;
		it=it->next;
		i++;
	}
	DEBUG(printf("%s:going to sort %d rec\n",__func__,ha->nrec));
	heapsort(s,ha->nrec,ord->ord);
	it=s[0].data;
	ha->first=ha->last=it;
	it->pre=it->next=NULL;
	for(i=1;i<ha->nrec;i++){
	    it=s[i].data;
		ha->last->next=it;
		it->pre=ha->last;
		it->next=NULL;
		ha->last=it;
	}
	free(s);
}
void heapsort(struct sortitem *s,int len,int ord){
    int i,j,k,p,cmp;
	if(len<=1) return;
	i=len>>1;
	while(i>0){
	    j=i;
		k=len>>1;
		while(j<=k){
		    p=j;
			j<<=1;
		    if(j<len&&(((cmp=strcomparestr(s[j-1].key,s[j].key))<0&&ord==ASC)||(cmp>0&&ord==DSC))) j++;
			if(((cmp=strcomparestr(s[p-1].key,s[j-1].key))>=0&&ord==ASC)||(cmp<=0&&ord==DSC)) break;
			//swap(j-1,p-1);
			struct sortitem tmp;
			tmp.key=s[j-1].key;
			tmp.data=s[j-1].data;
			s[j-1].data=s[p-1].data;
			s[j-1].key=s[p-1].key;
			s[p-1].data=tmp.data;
			s[p-1].key=tmp.key;
		}
		i--;
	}
	i=len;
	while(i>1){
	    i--;
	    //SWAP(a[0],a[i]);
		struct sortitem tmp;
		tmp.key=s[0].key;
		tmp.data=s[0].data;
		s[0].key=s[i].key;
		s[0].data=s[i].data;
		s[i].key=tmp.key;
		s[i].data=tmp.data;
		j=1;
		k=i>>1;
		while(j<=k){
		    p=j;
			j<<=1;
			if(j<i&&(((cmp=strcomparestr(s[j-1].key,s[j].key))<0&&ord==ASC)||(cmp>0&&ord==DSC))) j++;
			if(((cmp=strcomparestr(s[j-1].key,s[p-1].key))>0&&ord==ASC)||(cmp<0&&ord==DSC)){
			    //SWAP(a[p-1],a[j-1]);
				tmp.key=s[p-1].key;
				tmp.data=s[p-1].data;
				s[p-1].key=s[j-1].key;
				s[p-1].data=s[j-1].data;
				s[j-1].key=tmp.key;
				s[j-1].data=tmp.data;
			}
			else break;
		}
	}
}
struct condition *condcreate(char *key,int ksiz,uint32 flag,char *ref,int rsiz){
    struct condition *c=malloc(sizeof(struct condition));
	c->key=key;
	c->ksiz=ksiz;
	c->flag=flag;
	c->ref=ref;
	c->rsiz=rsiz;
	return c;
}
int indexadd(struct db *gdb,char *col,int csiz){
    struct bdb *idb;
	struct bparam bp;
	memset(&bp,0,sizeof(struct bparam));
	bp.minkey=2;
	bp.pgsiz=512;
	bp.align=4;
	bp.pcache_maxbytes=1024*32;
	bp.lcache_maxbytes=1024*32;
	bp.pcache_nbucket=bp.lcache_nbucket=31;
	pthread_rwlock_wrlock(&gdb->rwlock);
	if(gdb->cdbs==NULL) gdb->cdbs=hashcreate(0);
	else if((idb=hashget(gdb->cdbs,col,csiz))!=NULL){
	    DEBUG(printf("already exsit!\n"));
		pthread_rwlock_unlock(&gdb->rwlock);
		return false;
	}
	if((idb=bopen(gdb->dname,col,gdb->seq++,gdb->flag|O_CREAT|O_TRUNC,&bp))==NULL){
	    printf("%s:create index fail\n",__func__);
		pthread_rwlock_unlock(&gdb->rwlock);
		return false;
	}
	if(gdb->hd->Nrec!=0){
	    struct iterator *iter=iter_create(gdb->hd);
		if(iter==NULL){
		    printf("iter create fail\n");
			pthread_rwlock_unlock(&gdb->rwlock);
			return false;
		}
		struct bstring *key;
		struct bstring *val;
		while(iter_next(iter,&key,&val)){
		    struct harray *ha=strtohash(val->array,val->siz);
			struct bstring *it;
			if((it=hashget(ha,col,csiz))!=NULL){
			    if(!bput(idb,it->array,it->siz,key->array,key->siz,0)){
				    printf("%s:put idx fail\n",__func__);
				}
			}
			strdel(key);
			strdel(val);
			hashdelstr(ha);
		}
	}
	DEBUG(printf("%s:create a bdb:%x\n",__func__,idb));
	if(!hashput(gdb->cdbs,col,csiz,idb)){
	    printf("register idx fail\n");
		exit(1);
	}
	pthread_rwlock_unlock(&gdb->rwlock);
	return true;
}
int indexerase(struct db *gdb,char *iname,int ilen){
    pthread_rwlock_wrlock(&gdb->rwlock);
	struct bdb *idb;
	if((idb=hashget(gdb->cdbs,iname,ilen))==NULL){
	    pthread_rwlock_unlock(&gdb->rwlock);
		return false;
	}
	hasherase(gdb->cdbs,iname,ilen);
	bdelete(idb);
	pthread_rwlock_unlock(&gdb->rwlock);
	return true;
}
int dbdelete(struct db *gdb){
    //pthread_mutex_lock(&gdb->lock);
	pthread_rwlock_wrlock(&gdb->rwlock);
	struct bdb *idb;
	struct item *it=gdb->cdbs->first;
	while(it!=NULL){
	    idb=it->val;
		if(idb!=NULL) bdelete(idb);
		it=it->next;
	}
	hdbdelete(gdb->hd);
	hashdel(gdb->cdbs,NULL);
	pthread_rwlock_unlock(&gdb->rwlock);
	pthread_rwlock_destroy(&gdb->rwlock);
	//pthread_mutex_unlock(&gdb->lock);
	//pthread_mutex_destroy(&gdb->lock);
	if(gdb->dname!=NULL) free(gdb->dname);
	free(gdb);
	return true;
}
int dberase(struct db* gdb,struct condition* c){
    struct harray *ha=query(gdb,c,0,0,NULL);
	int ret=true;
	pthread_rwlock_rdlock(&gdb->rwlock);
	if(ha!=NULL&&ha->nrec!=0){
	    struct item *it=ha->first;
		while(it!=NULL){
		    struct bdb *idb;
		    struct harray *rec=it->val;
			struct item *ri=rec->first;
			while(ri!=NULL){
			    if((idb=hashget(gdb->cdbs,ri->key->array,ri->key->siz))!=NULL){
				    struct bstring *val=ri->val;
					if(!berase(idb,val->array,val->siz,it->key->array,it->key->siz)){
					    printf("%s:del from index fail\n",__func__);
						ret=false;
					}
				}
				ri=ri->next;
			}
			if(!hdberase(gdb->hd,it->key->array,it->key->siz)){
			    printf("%s:del from data file fail\n",__func__);
				ret=false;
			}
			it=it->next;
		}
	}
	hashdel(ha,(void(*)(void*))hashdelstr);
	pthread_rwlock_unlock(&gdb->rwlock);
	return ret;
}
int dberasebyid(struct db* gdb,char *id,int idsiz){
    struct bstring *key=strcreate(id,idsiz);
	struct bstring *val=hdbgetstr(gdb->hd,key);
	if(val==NULL){
	    strdel(key);
	    return false;
	}
	struct harray *h=strtohash(val->array,val->siz);
	struct item *it=h->first;
	int ret=true;
	pthread_rwlock_rdlock(&gdb->rwlock);
	while(it!=NULL){
	    struct bdb *idb=hashget(gdb->cdbs,it->key->array,it->key->siz);
		if(idb!=NULL&&!berase(idb,((struct bstring*)(it->val))->array,((struct bstring*)(it->val))->siz,id,idsiz)){
		    printf("%s:error del index\n",__func__);
			ret=false;
		}
	}
	pthread_rwlock_unlock(&gdb->rwlock);
	if(!hdberase(gdb->hd,id,idsiz)){
	    printf("%s:error del data\n",__func__);
		ret=false;
	}
	strdel(key);
	strdel(val);
	hashdelstr(h);
	return ret;
}
int trans_beg(struct db *gdb){
    struct bdb* idb=hashfirst(gdb->cdbs,NULL);
	int ret;
	while(idb!=NULL){
	    ret=bdbtrans_beg(idb);
		if(!ret){
		    idb=hashpre(gdb->cdbs,NULL);
			while(idb!=NULL){
			    bdbtrans_cancel(idb);
				idb=hashpre(gdb->cdbs,NULL);
			}
			return false;
		}
		idb=hashnext(gdb->cdbs,NULL);
	}
	hdbtrans_beg(gdb->hd);
}
int trans_commit(struct db *gdb){
    hdbtrans_commit(gdb->hd);
    struct bdb* idb=hashfirst(gdb->cdbs,NULL);
	while(idb!=NULL){
	    bdbtrans_commit(idb);
		idb=hashnext(gdb->cdbs,NULL);
	}
	return true;
}
int trans_abort(struct db *gdb){
    struct bdb *idb=hashfirst(gdb->cdbs,NULL);
	while(idb!=NULL){
	    bdbtrans_abort(idb);
		idb=hashnext(gdb->cdbs,NULL);
	}
	hdbtrans_abort(gdb->hd);
	return true;
}
struct harray *dbinter_next(struct iterator *iter){
    struct bstring *key=NULL,*val=NULL;
	struct harray *ha=NULL;
	if(iter_next(iter,&key,&val)&&key!=NULL){
		ha=strtohash(STRVAL(val),STRLEN(val));
		hashput(ha,"id",2,key);
	}
	return ha;
}
